package com.sanerzone.smscenter.biz.listener;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.google.common.collect.Maps;
import com.sanerzone.common.support.sequence.MsgId;
import com.sanerzone.common.support.utils.FstObjectSerializeUtil;
import com.sanerzone.smscenter.biz.mapper.SmsSendMapper;
import com.sanerzone.smscenter.biz.message.SMSRTMessage;
import com.sanerzone.smscenter.biz.utils.TableNameHelper;
import com.sanerzone.smscenter.common.tools.MQHelper;
import com.sanerzone.smscenter.common.tools.SpringContextHelper;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class UPSMSRTListener implements MessageListenerConcurrently {

	public static Logger logger = LoggerFactory.getLogger(UPSMSRTListener.class);
	@Autowired
	private MQHelper mqHelper;


	@Autowired
	private SmsSendMapper smsSendMapper;

	public static String getTableIndex(String bizid) {
		if (StringUtils.isNotBlank(bizid)) {
			return String.valueOf(new MsgId(bizid).getDay());
		} else {
			return "";
		}
	}
	@Override
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
		SqlSession sqlSession = SpringContextHelper.getBean(SqlSessionFactory.class).openSession(ExecutorType.SIMPLE, false);
		try {

			int index = 0;
			SMSRTMessage smsRTMessage = null;
			for (MessageExt msg : msgs) {

				smsRTMessage = (SMSRTMessage) FstObjectSerializeUtil.read(msg.getBody());
				String phone1=smsRTMessage.getPhone();
				if(smsRTMessage == null || smsRTMessage.getSmsMt() == null){
					logger.error("err回调缓存取mt为空,手机号{},流水号{}",phone1,smsRTMessage.getMsgid());
					continue;
				}

				if (msg.getReconsumeTimes()==3){
					continue;
				}
				String gwID=smsRTMessage.getGateWayID();

				logger.info( "正在回调状态中:{}:{}",phone1,gwID );
				//更新状态
				Map<String,Object> map = Maps.newHashMap();
				map.put("mtBatch", smsRTMessage.getSmsMt().getMtBatchNum());
				if("DELIVRD".equals(smsRTMessage.getStat())){//成功的情况
					map.put("gatewayStatus", "T100");
				}else{
					map.put("gatewayStatus", "F2" + smsRTMessage.getStat());//失败的情况
//					//tag key随便写
//					mqHelper.sendSmsRC( "1","1",smsRTMessage );

					String mtBatchNum=smsRTMessage.getSmsMt().getMtBatchNum();
					String phone=smsRTMessage.getPhone();
					String tableIndex=getTableIndex( mtBatchNum );
					String tableName="sms_send_" + tableIndex;
					Map map1=new HashMap( );
					map1.put( "mtBatch",mtBatchNum );
					map1.put( "phone",phone );
					map1.put( "tableName", tableName);
					//这里无法定位，可能会取到多条记录
//					List<SmsSend> smsSends=smsSendMapper.getByMtBatchNumAndPhone( map1 );

//					boolean canUpdate=false;
//					for (int i=0;i<smsSends.size();i++){
//						SmsSend smsSend=smsSends.get( i );
//						String gatewayStatus=smsSend.getGatewayStatus();
//						if (gatewayStatus!=null&&gatewayStatus.startsWith( "P" )){
//							canUpdate=true;
//							break;
//						}
//					}
//					if (canUpdate==false){
//						logger.info( "更新状态数据库无法查找到,mq稍后重新消费,警告消费3次" );
//						//这里先这么处理，因为重复号码无法定位
//						return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//					}else{
//						//反充
//						//反充任务开始
//						String userId=smsRTMessage.getSmsMt().getSmsREQMessage().getUserid();
//						String amountKey = AccountCacheHelper.getAmountKey("sms", userId);
//						String content=smsRTMessage.getSmsMt().getSmsContent();
//						int smsSize = StringHelper.smsSize(content);
//						JedisUtils.incrBy( amountKey,smsSize );
//						logger.info( "回调状态结果为失败,进行反充:用户id:{},反充结束,反充条数为{}",userId,smsSize );
//					}
				}

				map.put("submitTime", smsRTMessage.getSmsMt().getSubmitTime());
				map.put("phone", smsRTMessage.getPhone());
                map.put("reportTime", smsRTMessage.getReportReceiveTime());
                map.put( "statusChangeTime",new Date( ) );
                map.put("tableName", TableNameHelper.getSmsSendTable(smsRTMessage.getSmsMt().getId()));
                //根据mt的批次和手机号码去定位一条明细
				int num = sqlSession.update("com.sanerzone.smscenter.biz.mapper.SmsSendMapper.batchUpdateGatewayStatus2", map);



				//				if (num == 0)
//				{
//					Map<String, Object> mapO = new HashMap<String, Object>();
//					mapO.put("tableName", "sms_report_sync");
//					mapO.put("data", smsRTMessage);
//					sqlSession.insert("com.sanerzone.smscenter.biz.mapper.SmsReportSyncMapper.insert", mapO);
//				}


				index++;


				if (index % 200 == 0) {
					sqlSession.commit();
				}
			}
			sqlSession.commit();

		} catch (Exception e) {
			logger.error("UPSMSRT-ERR", e);
		} finally {
			if (sqlSession != null) {
				sqlSession.close();
			}
		}
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}

}
